package org.jgroups.protocols;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.Condition;

import org.jgroups.Address;
import org.jgroups.Event;
import org.jgroups.Message;
import org.jgroups.View;
import org.jgroups.conf.PropertyConverters;
import org.jgroups.stack.Protocol;
import org.jgroups.annotations.ManagedAttribute;
import org.jgroups.annotations.Property;
import org.jgroups.annotations.GuardedBy;
import org.jgroups.util.Util;

/*
 * mickey hsieh
 * try to provide a upper limit time cap for each package
 */
public class FC2 extends Protocol {

	/*
	 * default max time out in mini seconds , match with size
	 * timOutList.length = time_out_size.lenght + 1
	 */
    private long[] time_out_list = { 19, 39,  59,  79,  99, 119, 139, 159 };
	/*
	 * number of package in FRAG2
	 */
	private int[] time_out_size = { 35, 70,  140,  280, 560,  1120, 2240  };

    public long[] getTime_out_list() {
        return time_out_list;
    }

    @Property(name="time_out_list",converter= PropertyConverters.LongArray.class)
    public void setTime_out_list(long[] time_out_list) {
        this.time_out_list = time_out_list;
    }
//    protected Vector members =new Vector<Address>(11);

    protected Address localAddr;

    protected final static FC.FcHeader REPLENISH_HDR=new FC.FcHeader(FC.FcHeader.REPLENISH);
    protected final static FC.FcHeader CREDIT_REQUEST_HDR=new FC.FcHeader(FC.FcHeader.CREDIT_REQUEST);
    protected final static String name="FC";


    /* -----------------------------------------    Properties     -------------------------------------------------- */

    /**
     * Max number of bytes to send per receiver until an ack must
     * be received before continuing sending
     */
    @ManagedAttribute(description="Max number of bytes to send per receiver until an ack must " +
                                   "be received before continuing sending",writable=true)
    @Property(description="Max number of bytes to send per receiver until an ack must be received to proceed. Default is 500000 bytes")
    protected long max_credits=500000;

    /**
     * Max time (in milliseconds) to block. If credit hasn't been received after max_block_time, we send
     * a REPLENISHMENT request to the members from which we expect credits. A value <= 0 means to
     * wait forever.
     */
    @ManagedAttribute(description="Max time (in milliseconds) to block",writable=true)
    @Property(description="Max time (in milliseconds) to block. Default is 5000 msec")
    protected long max_block_time=5000;


    /** Keeps track of the end time after which a message should not get blocked anymore */
    private static final ThreadLocal<Long> end_time=new ThreadLocal<Long>();


    /**
     * If credits fall below this limit, we send more credits to the sender. (We also send when
     * credits are exhausted (0 credits left))
     */
    @ManagedAttribute(description="If credits fall below this limit, we send more credits to the sender",writable=true)
    @Property(description="If credits fall below this limit, we send more credits to the sender. Default is 0.25")
    private double min_threshold=0.20;

    /**
     * Computed as <tt>max_credits</tt> times <tt>min_theshold</tt>. If explicitly set, this will
     * override the above computation
     */
    @ManagedAttribute(description="Computed as max_credits x min_theshold",writable=true)
    @Property(description="Computed as max_credits x min_theshold unless explicitely set")
    private long min_credits=0;

    /**
     * Whether an up thread that comes back down should be allowed to
     * bypass blocking if all credits are exhausted. Avoids JGRP-465.
     * Set to false by default in 2.5 because we have OOB messages for credit replenishments - this flag should not be set
     * to true if the concurrent stack is used
     */
    @Property
    protected boolean ignore_synchronous_response=false;




    /* ---------------------------------------------   JMX      ------------------------------------------------------ */


    protected int num_blockings=0;
    protected int num_credit_requests_received=0, num_credit_requests_sent=0;
    protected int num_credit_responses_sent=0, num_credit_responses_received=0;
    protected long total_time_blocking=0;




    /* --------------------------------------------- Fields ------------------------------------------------------ */


    /**
     * List of members from whom we expect credits
     */
    @GuardedBy("sent_lock")
    protected final Set<Address> creditors=new HashSet<Address>(11);


    /**
     * Whether FC is still running, this is set to false when the protocol terminates (on stop())
     */
    protected volatile boolean running=true;


    private boolean frag_size_received=false;


    /**
     * the lowest credits of any destination (sent_msgs)
     */
    @GuardedBy("sent_lock")
    protected long lowest_credit=max_credits;

    /** Lock protecting sent credits table and some other vars (creditors for example) */
    protected final Lock sent_lock=new ReentrantLock();

    /**
     * Map<Address,Long>: keys are members, values are credits left (in bytes).
     * For each receive, the credits for the sender are decremented by the size of the received message.
     * When the credits are 0, we refill and send a CREDIT message to the sender. Sender blocks until CREDIT
     * is received after reaching <tt>min_credits</tt> credits.
     */
    @GuardedBy("received_lock")
    protected final Map<Address, Long> received=new ConcurrentHashMap<Address, Long>(11);
    protected int receivedSize;
    /** Lock protecting received credits table */
    protected final Lock received_lock=new ReentrantLock();


    /** Mutex to block on down() */
    protected final Condition credits_available= sent_lock.newCondition();


    /**
     * Thread that carries messages through up() and shouldn't be blocked
     * in down() if ignore_synchronous_response==true. JGRP-465.
     */
    protected Thread ignore_thread;

    /** Last time a credit request was sent. Used to prevent credit request storms */
    @GuardedBy("sent_lock")
    protected long last_credit_request=0;

    protected int creditorSize ;

    public final String getName() {
        return name;
    }

    public void init() throws Exception {
        boolean min_credits_set = min_credits != 0;
        if(!min_credits_set)
            min_credits=(long)(max_credits * min_threshold);

    }

    public void start() throws Exception {
        super.start();
        if(!frag_size_received) {
            log.warn("No fragmentation protocol was found. When flow control (e.g. FC or SFC) is used, we recommend " +
                    "a fragmentation protocol, due to http://jira.jboss.com/jira/browse/JGRP-590");
        }

        sent_lock.lock();
        try {
            running=true;

            lowest_credit=max_credits;
        }
        finally {
            sent_lock.unlock();
        }
    }

    public void stop() {
        super.stop();
        sent_lock.lock();
        try {
            running=false;
            ignore_thread=null;
            credits_available.signalAll(); // notify all threads waiting on the mutex that we are done
        }
        finally {
            sent_lock.unlock();
        }
    }



    @Override
    public Object up(Event evt) {
        switch(evt.getType()) {

            case Event.MSG:

                // JGRP-465. We only deal with msgs to avoid having to use a concurrent collection; ignore views,
                // suspicions, etc which can come up on unusual threads.
                if(ignore_thread == null && ignore_synchronous_response)
                    ignore_thread=Thread.currentThread();

                Message msg=(Message)evt.getArg();
                FC.FcHeader hdr=(FC.FcHeader)msg.getHeader(name);
                if(hdr != null) {
                    //if( log.isInfoEnabled() ) log.info("FC "+ hdr.toString()+ msg.getSrc().toString() );
                    switch(hdr.type) {
                        case FC.FcHeader.REPLENISH:
                            num_credit_responses_received++;
                            handleCredit(msg.getSrc(), (Number)msg.getObject());
                            break;
                        case FC.FcHeader.CREDIT_REQUEST:
                            Address sender=msg.getSrc();
                            Long sent_credits=(Long)msg.getObject();
                            handleCreditRequest(sender, sent_credits);
                            break;
                        default:
                            log.error("header type " + hdr.type + " not known");
                            break;
                    }
                    return null; // don't pass message up
                }
                else {
                    Address sender=msg.getSrc();
                    long new_credits=adjustCredit(received, received_lock, sender, msg.getLength());
                    try {
                        return up_prot.up(evt);
                    }
                    finally {
                        if(new_credits > 0) {
                            sendCredit(sender, new_credits);
                        }
                    }
                }

            case Event.VIEW_CHANGE:
                handleViewChange(((View)evt.getArg()).getMembers());
                break;

            case Event.SET_LOCAL_ADDRESS:
                localAddr=(Address)evt.getArg();
                break;

            case Event.CONFIG:
                Map<String,Object> map=(Map<String,Object>)evt.getArg();
                handleConfigEvent(map);
                break;
        }
        return up_prot.up(evt);
    }


    /**
     * Check whether sender has enough credits left. If not, send him some more
     * @param map The hashmap to use
     * @param lock The lock which can be used to lock map
     * @param sender The address of the sender
     * @param length The number of bytes received by this message. We don't care about the size of the headers for
     * the purpose of flow control
     * @return long Number of credits to be sent. Greater than 0 if credits needs to be sent, 0 otherwise
     */
    protected long adjustCredit(Map<Address,Long> map, final Lock lock, Address sender, int length) {
        if(sender == null) {
            if(log.isErrorEnabled()) log.error("src is null");
            return 0;
        }

        if(length == 0)
            return 0; // no effect

        lock.lock();
        try {
            long remaining_cred=decrementCredit(map, sender, length);
            if(log.isTraceEnabled()) log.trace("sender " + sender + " minus " + length
						+ " credits, " + remaining_cred + " remaining");
            if(remaining_cred == -1)
                return 0;
            long credit_response=max_credits - remaining_cred;
            if(credit_response >= min_credits) {
                map.put(sender, max_credits);
                return credit_response; // this will trigger sending of new credits as we have received more than min_credits bytes from src
            }
        }
        finally {
            lock.unlock();
        }
        return 0;
    }

    /**
     * Decrements credits from a single member, or all members in sent_msgs, depending on whether it is a multicast
     * or unicast message. No need to acquire mutex (must already be held when this method is called)
     * eliminate the multicast case to improve the performance
     * @param dest
     * @param credits
     * @return The lowest number of credits left, or -1 if a unicast member was not found
     */
    protected long decrementCredit(Map<Address, Long> m, Address dest, long credits) {
        long lowest=max_credits, new_credit;
        Long val;

//        boolean multicast=dest == null || dest.isMulticastAddress();
//        if(multicast) {
//            if(m.isEmpty())
//                return -1;
//            for(Map.Entry<Address, Long> entry: m.entrySet()) {
//                val=entry.getValue();
//                new_credit=val - credits;
//                entry.setValue(new_credit);
//                lowest=Math.min(new_credit, lowest);
//            }
//            return lowest;
//        }
            val=m.get(dest);
            if(val != null) {
                lowest=val;
                lowest-=credits;
                m.put(dest, lowest);
                return lowest;
            }
            else
                return -1;
    }

    protected void sendCredit(Address dest, long credit) {
//        if(log.isInfoEnabled())
//            log.info("replentished " + dest + " with " + credit	+ " credits");
        Number number;
        if(credit < Integer.MAX_VALUE)
            number=(int)credit;
        else
            number=credit;
        Message msg=new Message(dest, null, number);
        msg.setFlag(Message.OOB);
        msg.putHeader(name, REPLENISH_HDR);
        down_prot.down(new Event(Event.MSG, msg));
        num_credit_responses_sent++;
    }


    @Override
    public Object down(Event evt) {
        switch(evt.getType()) {
            case Event.MSG:
                return handleDownMessage(evt);
            case Event.CONFIG:
                handleConfigEvent((Map<String,Object>)evt.getArg());
                break;
            case Event.SET_LOCAL_ADDRESS:
                localAddr=(Address)evt.getArg();
                break;
            case Event.VIEW_CHANGE:
                handleViewChange(((View)evt.getArg()).getMembers());
                break;
        }
        return down_prot.down(evt); // this could potentially use the lower protocol's thread which may block
    }



	/**
	 * @return timeout value
	 */
	private long getTimeOut(int size) {
		int i = 0 ;
		for ( ; i < time_out_size.length ; i ++  ) {
			if ( size < time_out_size[i] )
				return time_out_list[i];
		}
		// return the last one
		return time_out_list[i];
	}

	private long getTimeOut(Message msg) {
		FragHeader3 hdr = (FragHeader3) msg.getHeader("FRAG3");
		if ( hdr == null )
			return time_out_list[0];
		else
			return getTimeOut(hdr.num_frags );
	}

    /**
     * @param sender The sender who requests credits
     * @param left_credits Number of bytes that the sender has left to send messages to us
     */
    protected void handleCreditRequest(Address sender, Long left_credits) {
        if(sender == null) return;
        Message msg=new Message(sender, localAddr, max_credits);
        msg.putHeader(name, REPLENISH_HDR);
        //msg.setFlag(Message.OOB);
        down_prot.down(new Event(Event.MSG, msg));
        num_credit_responses_sent++;
    }



    protected Object handleDownMessage(Event evt) {
        Message msg=(Message)evt.getArg();
        int length=msg.getLength();
        //Address dest=msg.getDest();

        sent_lock.lock();
        try {
            if(length > lowest_credit) { // then block and loop asking for credits until timeout or enough creidtor response
                if(ignore_synchronous_response && ignore_thread == Thread.currentThread()) { // JGRP-465
                    if(log.isTraceEnabled())
                        log.trace("bypassing blocking to avoid deadlocking " + Thread.currentThread());
                }
                else {
                    //long start_blocking=System.currentTimeMillis();
                    long maxTimeout = getTimeOut( msg );
                	boolean rc = false ;
                    while(length > lowest_credit && running) {
                        try {
                            rc = credits_available.await(maxTimeout, TimeUnit.MILLISECONDS);
                            num_blockings++; // we count overall blockings, not blockings for *all* threads
                            long reset = 0 ;
                            //int receivedSize = received.size() ;
                            if(rc || !running ) {
                                // do nothing note reset lowest_credits, creditorSize and credditors.remove was performed
                                // by handlCredit credits_available.signal()
                                if( creditors.size() > 0 ) {
                                    creditors.removeAll( creditors);
                                }
                            }
                            else {  // when it did not received from all
                                creditorSize = creditors.size() ;
                                if ( receivedSize == 0)
                                   reset = max_credits ;
                                else if ( creditorSize == 0 ) {
                                    sendCreditRequestAll(max_credits, msg.getDest() );
                                    reset = max_credits /4 ;
                                }
                                else if ( creditorSize >= receivedSize)
                                    reset = max_credits ;
                                else {
                                     reset = max_credits * creditorSize / receivedSize;
                                     if ( reset < max_credits /4 ) {
                                         reset = max_credits /4;
                                     }
                                }
                                lowest_credit = reset ;
                                creditors.removeAll( creditors);
                                // else it was received after it was time out
                            }
                            if( log.isInfoEnabled()) log.info("rc "+rc +" "+num_blockings+ " len "+ length +" lowest_credit "+ lowest_credit
                                    +" reset "+ reset + " crditor "+ creditorSize +" recv "+ receivedSize);
                            creditorSize = 0;
                            // must get out loop
                            break;
                        }
                        catch(InterruptedException e) {
                            // set the interrupted flag again, so the caller's thread can handle the interrupt as well
                            // bela June 15 2007: don't do this as this will trigger an infinite loop !!
                            // (http://jira.jboss.com/jira/browse/JGRP-536)
                            // Thread.currentThread().interrupt();
                        }
                    } //while
                } //else
            }  // length > lowest_credit
            else  //length <= lowest_credit
                lowest_credit = lowest_credit - length ;
        } //try
        finally {
            sent_lock.unlock();
        }
        // send message - either after regular processing, or after blocking (when enough credits available again)
        return down_prot.down(evt);
	}

    /**
     * send a multicast credit request, not a unitcast. It is much more scalable
     * we don't want to track each receiver
     * @param credits_left The number of bytes (of credits) left for dest
     */
    protected void sendCreditRequestAll(Long credits_left, Address dest) {
        if ( dest != null && dest.equals(localAddr) )
            return ;
        Message msg=new Message(dest, localAddr, credits_left);
        msg.putHeader(name, CREDIT_REQUEST_HDR);
        down_prot.down(new Event(Event.MSG, msg));
        num_credit_requests_sent++;
    }



    protected void handleViewChange(Vector<Address> mbrs) {
        Address addr;
        if(mbrs == null) return;

        received_lock.lock();
        try {
            // add members not in membership to received and sent hashmap (with full credits)
            for(int i=0; i < mbrs.size(); i++) {
                addr=mbrs.elementAt(i);
                if(!received.containsKey(addr))
                    received.put(addr, max_credits);
            }
            // remove members that left
            for(Iterator<Address> it=received.keySet().iterator(); it.hasNext();) {
                addr=it.next();
                if(!mbrs.contains(addr))
                    it.remove();
            }

            // fixed http://jira.jboss.com/jira/browse/JGRP-754 (CCME)
            for(Iterator<Address> it=creditors.iterator(); it.hasNext();) {
                Address creditor=it.next();
                if(!mbrs.contains(creditor))
                    it.remove();
            }
            this.receivedSize = received.size() ;

        }
        finally {
            received_lock.unlock();
        }

    }



    protected void handleCredit(Address sender, Number increase) {
        if(sender == null) return;

        sent_lock.lock();
        try {
            creditors.add( sender);
            // make sure it is receivedSize
            if(creditors.size() >= receivedSize ) { // && !was_empty) {
                creditorSize = creditors.size();
                creditors.removeAll(creditors);
                lowest_credit = max_credits;
                credits_available.signal();
                }
            }
        finally {
            sent_lock.unlock();
        }
    }

    protected void handleConfigEvent(Map<String,Object> info) {
        if(info != null) {
            Integer frag_size=(Integer)info.get("frag_size");
            if(frag_size != null) {
                if(frag_size > max_credits) {
                    log.warn("The fragmentation size of the fragmentation protocol is " + frag_size +
                            ", which is greater than the max credits. While this is not incorrect, " +
                            "it may lead to long blockings. Frag size should be less than max_credits " +
                            "(http://jira.jboss.com/jira/browse/JGRP-590)");
                }
                frag_size_received=true;
            }
        }
    }




	public int[] getTime_out_size() {
		return time_out_size;
	}

    @Property(name="time_out_size",converter= PropertyConverters.LongArray.class)
	public void setTime_out_size(int[] time_out_size) {
		this.time_out_size = time_out_size;
	}
}
